/*
 * @Author: liuzelin
 * @Date: 2023-09-11 21:33:12
 * @LastEditors: liuzelin
 * @LastEditTime: 2024-05-12 21:29:08
 * @Description: file content
 */
#include "endpoint.hpp"

#include <map>
#include <memory>

#include "panda-framework/panda/panda.hpp"
#include "protocol/expand.hpp"
#include "protocol/frame.hpp"
#include "utils/histogram.hpp"
#include "utils/sack_generator.hpp"
#include "utils/seq_sort.hpp"

using namespace Panda;
using namespace PandaUtils;

namespace {
constexpr uint32_t kQuantile = 1041529569;  // 0.97 in Q30. 1<<30 * 0.97
constexpr int forget_factor = 32604;        // 0.995 Q15
constexpr double start_forget_weight = 2;
constexpr size_t sack_interval_large_packet = 50;
constexpr int64_t sack_min_interval = 10;
constexpr int64_t sack_max_interval = 50;
constexpr int64_t sack_max_size = 1000;
constexpr int64_t keep_alive_interval = 2000;
constexpr int64_t keep_alive_interval_on_time_out = 200;
constexpr int64_t time_out_ms = 30000;
constexpr uint8_t pos_start = 0x2;
constexpr uint8_t pos_end = 0x1;
uint32_t NowUint32() { return Time::NowMs(); }
const uint8_t version = 0;
}  // namespace

struct Block {
  Frame frame;
  union {
    bool acked = false;
    bool called;
  };
  union {
    int64_t retrans_count = 0;
    int64_t recv_time_ms;
  };
};
typedef std::map<uint16_t, std::unique_ptr<Block>, SeqSortWarper<uint16_t>>
    BlockCache;
typedef BlockCache::iterator BlockIterm;
typedef BlockCache::reverse_iterator BlockRIterm;

size_t GetCompleteFrame(BlockCache &cache, std::optional<uint16_t> seq,
                        std::vector<BlockIterm> &blocks) {
  if (cache.empty()) return 0;
  size_t frame_size = 0;
  bool complete = false;
  if (seq != std::nullopt) {
    BlockIterm it = cache.find(*seq);
    uint16_t seq = it->first;
    BlockIterm start = cache.end();
    for (BlockIterm pos = it;; seq--,--pos) {
      if (seq != pos->first) break;
      if ((pos->second->frame.header.pos & pos_start) &&
          (pos->second->frame.header.pos & pos_end) && pos->first == seq) {
        start = pos;
        break;
      }
      if ((pos->second->frame.header.pos & pos_end) && pos->first != seq) break;
      if (pos->second->frame.header.pos & pos_start) {
        start = pos;
        break;
      }
      if (pos == cache.begin()) break;
    }
    if (start != cache.end()) {
      blocks.emplace_back(start);
      seq = start->first;
      for (BlockIterm pos = start; pos != cache.end();seq++, ++pos) {
        if (seq != pos->first) break;
        frame_size += pos->second->frame.payload_len;
        if (pos->second->frame.header.pos & pos_end) {
          blocks.emplace_back(pos);
          complete = true;
          break;
        }
      }
    }
  } else {
    uint16_t seq = cache.begin()->first;
    bool read_start = false;
    for (BlockIterm it = cache.begin(); it != cache.end(); seq++,++it) {
      if (seq != it->first) break;
      frame_size += it->second->frame.payload_len;
      if (it->second->frame.header.pos & pos_start) {
        read_start = true;
        blocks.emplace_back(it);
      }
      if (it->second->frame.header.pos & pos_end) {
        if (read_start) {
          complete = true;
          blocks.emplace_back(it);
        }
        break;
      }
    }
  }
  if (complete)
    return frame_size;
  else
    return 0;
}

bool CalculateRtt(FrameType type) {
  if (type == FrameType::KeepAliveAck || type == FrameType::ShakeHandsAck ||
      type == FrameType::Sack) {
    return true;
  }
  return false;
}

struct JitterEstimator {
  JitterEstimator()
      : jitter_histogram(1000, forget_factor, start_forget_weight) {}
  Histogram jitter_histogram;
  int64_t last_recv_ms = 0;
  uint32_t last_packet_ts = 0;
  uint32_t max_packet_ts = 0;
  int64_t max_packet_ts_ms = 0;
  int last_jitter = 0;
};

struct Session {
  enum Status {
    None,
    Connecting,
    Connected,
  } status;
  explicit Session(const SessionParam &param) : param(param) {
    status = Session::Status::None;
    send_seq = 0;
    recv_seq = 0;
    configure.is_order = param.is_order;
    configure.is_reliable = param.is_reliable;
    configure.enable_time_out_retrans = param.enable_time_out_retrans;
    ByteReaderWriter<uint16_t>::WriteByte(configure.max_retrans_num,
                                          param.max_retrans_num);
    ByteReaderWriter<uint16_t>::WriteByte(configure.max_retrans_time,
                                          param.max_retrans_time);
    ByteReaderWriter<uint32_t>::WriteByte(configure.sid, param.sid);
  }
  explicit Session(const struct SessionConfigure *config) {
    memcpy(&configure, config, sizeof(struct SessionConfigure));
    status = Session::Status::None;
    send_seq = 0;
    recv_seq = 0;
    param.is_order = configure.is_order;
    param.is_reliable = configure.is_reliable;
    param.enable_time_out_retrans = configure.enable_time_out_retrans;
    param.sid = ByteReaderWriter<uint32_t>::ReadByte(this->configure.sid);
    param.max_retrans_time =
        ByteReaderWriter<uint16_t>::ReadByte(this->configure.max_retrans_time);
    param.max_retrans_num =
        ByteReaderWriter<uint16_t>::ReadByte(this->configure.max_retrans_num);
    param.max_cache_size = 5 * 1024 * 1024;
  }
  SessionParam param;
  struct SessionConfigure configure;
  BlockCache send_cache;
  size_t send_cache_size = 0;
  BlockCache recv_cache;
  uint16_t send_seq = 0;
  uint16_t recv_seq = 0;
  uint16_t last_recv_sack_seq = 0;
  SACKGenerator sack_generator;
  int64_t last_sack_send_ms = 0;
  size_t recv_data_num_since_last_sack = 0;
};
class EndpointImpl : public EndpointInterface, public UDPSocketListener {
 public:
  explicit EndpointImpl(const EndpointConfigure &configure);
  ~EndpointImpl();
  void Bind(const char *ip, uint16_t port) override;
  void Connect(const char *ip, uint16_t port) override;
  const SocketAddress &GetLocal() override { return this->socket->GetLocal(); }
  const SocketAddress &GetRemote() override { return *remoteAddr; }
  size_t Send(uint8_t *data, size_t len, const SessionParam &param) override;
  void Close() override;

 protected:
  void OnReadFrom(char *data, size_t len, const SocketAddress &from) override;
  void OnSend(char *data, size_t len, int status, const char *msg) override;

 private:
  void InnerCheckFrameComplete(Session &session,
                               std::optional<uint16_t> seq = std::nullopt);
  bool InnerCallBackCompleteFrame(Session &session,
                                  std::optional<uint16_t> seq = std::nullopt);
  void InnerCompleteFrameSuccess(Session &session,
                                 std::vector<BlockIterm> &frame);
  bool VerifyStatus(FrameType type);
  void InnerSendTo(const uint8_t *data, size_t len,
                   const SocketAddress &remote);
  void InnerSend(Session &session);
  void InnerRecv(Session &session, char *data, size_t len);
  void InnerRecvSack(Session &session, char *data, size_t len);
  void InnerSendSack(Session &session);
  void InnerTimer();
  void InnerSendControlFrame(FrameType type, uint32_t ts = NowUint32());
  void InnerSendDisconnect();
  void InnerSendBlock(Session &session, Block *block);
  void InnerUpdateRecvJitter(uint32_t ts, int64_t now);
  void InnerRecvShakeHands(FrameType type, const SocketAddress &from,
                           int64_t now, uint32_t ts);
  void InnerRecvKeepAlive(FrameType type, int64_t now, uint32_t ts);
  void InnerRecvDisConnect(FrameType type, int64_t now, uint32_t ts);
  enum Status {
    Init,
    Connecting,
    Connected,
    TimeOut,
    DisConnecting,
    DisConnected,
  };
  Status status = Status::Init;
  EndpointConfigure configure;
  Thread *workerThread = nullptr;
  UDPSocket *socket = nullptr;
  std::optional<SocketAddress> remoteAddr;
  std::map<uint32_t, Session> session;
  int64_t rtt = 0;
  int64_t last_recv_ack_ms = 0;
  SeqStatistician<uint32_t> globalSeqStatistician;
  NetworkSpeedStatistics networkSpeedStatistics;
  JitterEstimator jitter_estimator;
  uint32_t globalSeq = 0;
  int64_t last_keepalive_send = 0;
};

EndpointInterface *EndpointInterface::CreateTransport(
    const EndpointConfigure &configure) {
  return new EndpointImpl(configure);
}

EndpointImpl::EndpointImpl(const EndpointConfigure &configure)
    : configure(configure), networkSpeedStatistics(100) {
  workerThread = ThreadManager::Instance()->CreateThread();
  workerThread->SetThreadName("TransportWorkerThread");
  workerThread->Start();
  socket = workerThread->CreateUDPSocket();
  socket->SetSocketListener(this);
  workerThread->PostDelayTask(
      PandaFromHere, [this]() { this->InnerTimer(); }, configure.ticket_ms);
}

EndpointImpl::~EndpointImpl() {
  workerThread->Invoke(PandaFromHere, [this]() {
    delete socket;
    socket = nullptr;
  });
  workerThread->Stop();
  delete workerThread;
}

void EndpointImpl::Bind(const char *ip, uint16_t port) {
  workerThread->Invoke(PandaFromHere, [&]() {
    SocketAddress addr(ip, port);
    socket->Bind(addr);
  });
}

void EndpointImpl::Connect(const char *ip, uint16_t port) {
  workerThread->Invoke(PandaFromHere, [&]() {
    PandaCheck(status == Status::Init);
    status = Status::Connecting;
    remoteAddr = SocketAddress(ip, port);
    InnerSendControlFrame(FrameType::ShakeHands);
    return;
  });
}

size_t EndpointImpl::Send(uint8_t *data, size_t len,
                          const SessionParam &param) {
  return workerThread->Invoke<size_t>(PandaFromHere, [&]() {
    if (status != Status ::Connected) {
      return (size_t)-2;
    }
    auto it = this->session.find(param.sid);
    if (it == this->session.end()) {
      it = this->session.emplace(std::make_pair(param.sid, Session(param)))
               .first;
    }

    Session &session = it->second;

    if (session.param.max_cache_size < len + session.send_cache_size) {
      return (size_t)-1;
    }
    size_t num =
        len / this->configure.mtu + (len % this->configure.mtu ? 1 : 0);
    uint16_t s_seq = session.send_seq;
    uint16_t e_seq = s_seq + num - 1;
    size_t off_set = 0;
    for (uint16_t seq = s_seq; seq <= e_seq; seq++) {
      std::unique_ptr<Block> block(new Block());
      block->frame.header.ver = version;
      block->frame.header.ext = 0;
      block->frame.header.padding = 0;
      block->frame.header.pos = 0;
      if (seq == s_seq) {
        block->frame.header.pos |= pos_start;
      }
      if (seq == e_seq) {
        block->frame.header.pos |= pos_end;
      }
      block->frame.header.type = FrameType::Data;
      ByteReaderWriter<uint32_t>::WriteByte(block->frame.header.sid, param.sid);
      ByteReaderWriter<uint16_t>::WriteByte(block->frame.header.seq, seq);
      block->frame.payload_len =
          seq == e_seq ? len % this->configure.mtu : this->configure.mtu;
      memcpy(block->frame.data, data + off_set, block->frame.payload_len);
      block->frame.len = block->frame.payload_len + sizeof(struct Header);
      block->retrans_count = 0;
      off_set += block->frame.payload_len;
      session.send_cache.emplace(std::make_pair(seq, std::move(block)));
    }
    session.send_cache_size += len;
    session.send_seq += num;
    InnerSend(session);
    return len;
  });
}

void EndpointImpl::InnerRecvShakeHands(FrameType type,
                                       const SocketAddress &from, int64_t now,
                                       uint32_t ts) {
  if (type == FrameType::ShakeHands) {
    switch (this->status) {
      case Status::Init: {
        this->status = Status::Connecting;
        this->remoteAddr = from;
        InnerSendControlFrame(FrameType::ShakeHands);
        break;
      }
      default:
        break;
    }
    LOG(Info) << "Transport:" << this
              << " Recv ShakeHands From:" << from.ToString();
    InnerSendControlFrame(FrameType::ShakeHandsAck, ts);
  } else if (type == FrameType::ShakeHandsAck) {
    switch (this->status) {
      case Status::Init:
      case Status::DisConnecting:
      case Status::DisConnected:
      case Status::Connected: {
        LOG(Info) << "Transport:" << this
                  << " Recv ShakeHandsAck From:" << from.ToString()
                  << " Status Error:" << this->status;
        // ERROR
        break;
      }
      case Status::Connecting: {
        LOG(Info) << "Transport:" << this
                  << " Recv ShakeHandsAck From:" << from.ToString()
                  << " Status Change Connected Rtt:"
                  << (uint32_t)((uint32_t)(now)-ts) << "ms";
        this->status = Status::Connected;
        this->configure.listener->OnConnect(this, this->remoteAddr->Ip(),
                                            this->remoteAddr->Port());
        break;
      }
      default:
        break;
    }
  }
}

void EndpointImpl::InnerRecvKeepAlive(FrameType type, int64_t now,
                                      uint32_t ts) {
  if (type == FrameType::KeepAlive) {
    if (this->status == Status::Init) {
      // ERROR
      return;
    }
    InnerSendControlFrame(FrameType::KeepAliveAck, ts);
  } else if (type == FrameType::KeepAliveAck) {
    this->last_recv_ack_ms = now;
    LOG(Info) << "Transport:" << this
              << " Recv KeepAliveAck Rtt:" << (uint32_t)((uint32_t)(now)-ts)
              << "ms";
  }
}

void EndpointImpl::InnerRecvDisConnect(FrameType type, int64_t now,
                                       uint32_t ts) {
  if (type == FrameType::DisConnect) {
    LOG(Info) << "Transport:" << this << " Recv DisConnect";
    if (this->status != Status::DisConnecting)
      this->status = Status::DisConnecting;
    InnerSendControlFrame(FrameType::DisConnect);
    InnerSendControlFrame(FrameType::DisConnectAck, ts);
  } else if (type == FrameType::DisConnectAck) {
    LOG(Info) << "Transport:" << this << " Recv DisConnectAck";
    this->status = Status::DisConnected;
    this->configure.listener->OnDisConnect(this);
  }
}

void EndpointImpl::InnerUpdateRecvJitter(uint32_t ts, int64_t now) {
  if (this->jitter_estimator.last_recv_ms == 0) {
    this->jitter_estimator.last_recv_ms = now;
    this->jitter_estimator.last_packet_ts = ts;
    this->jitter_estimator.max_packet_ts = ts;
    this->jitter_estimator.max_packet_ts_ms = now;
  } else {
    int jitter_now = now - this->jitter_estimator.last_recv_ms -
                     (ts - this->jitter_estimator.last_packet_ts);
    int jitter = jitter_now > this->jitter_estimator.last_jitter
                     ? jitter_now - this->jitter_estimator.last_jitter
                     : this->jitter_estimator.last_jitter - jitter_now;
    this->jitter_estimator.last_jitter = jitter_now;
    this->jitter_estimator.jitter_histogram.Add(
        std::min(std::abs(jitter), 999));
    this->jitter_estimator.last_recv_ms = now;
    this->jitter_estimator.last_packet_ts = ts;
    if (SeqSortWarper<uint32_t>()(ts, this->jitter_estimator.max_packet_ts)) {
      this->jitter_estimator.max_packet_ts = ts;
      this->jitter_estimator.max_packet_ts_ms = now;
    }
    LOG(Debug) << "Transport:" << this << " Recv Jitter:" << jitter
               << " JitterGain:"
               << this->jitter_estimator.jitter_histogram.Quantile(kQuantile)
               << " TimeStamp:" << ts
               << " MaxTimeStamp:" << this->jitter_estimator.max_packet_ts;
  }
}

void EndpointImpl::InnerSendBlock(Session &session, Block *block) {
  ByteReaderWriter<uint32_t>::WriteByte(block->frame.header.g_seq,
                                        this->globalSeq++);
  if (block->frame.header.type != FrameType::Sack)
    ByteReaderWriter<uint32_t>::WriteByte(block->frame.header.ts, NowUint32());
  this->InnerSendTo(block->frame.header_data, block->frame.len,
                    *this->remoteAddr);
  if (block->frame.header.type == FrameType::Data &&
      session.status == Session::Status::Connected) {
    block->frame.header.type == FrameType::ReTransData;
  }
}

void EndpointImpl::InnerCompleteFrameSuccess(Session &session,
                                             std::vector<BlockIterm> &frame) {
  if (session.param.is_order) {
    session.recv_seq = frame[1]->first + 1;
    session.recv_cache.erase(session.recv_cache.begin(), frame[1]);
    LOG(Debug) << "Session:" << session.param.sid
               << " Clear To:" << session.recv_seq;
  } else if (session.param.is_reliable) {
  } else {
  }
}

bool EndpointImpl::InnerCallBackCompleteFrame(Session &session,
                                              std::optional<uint16_t> seq) {
  std::vector<BlockIterm> blocks;
  size_t buffer_size = GetCompleteFrame(session.recv_cache, seq, blocks);
  if (buffer_size) {
    LOG(Debug) << "Session:" << session.param.sid
               << " Frame:" << blocks[0]->first << "-" << blocks[1]->first
               << " Size:" << buffer_size << " Complete";
    uint8_t *buffer = new uint8_t[buffer_size];
    size_t off_set = 0;
    for (BlockIterm it = blocks[0];; ++it) {
      PandaCheck(it->second->called == false)
          << " Session:" << session.param.sid << " Seq:" << it->first
          << "already called";
      memcpy(buffer + off_set,
             it->second->frame.data +
                 (it->second->frame.len - it->second->frame.payload_len -
                  sizeof(Header)),
             it->second->frame.payload_len);
      off_set += it->second->frame.payload_len;
      it->second->called = true;
      if (it == blocks[1]) break;
    }
    if (this->configure.listener)
      this->configure.listener->OnRecv(this, buffer, buffer_size,
                                       session.param.sid);
    InnerCompleteFrameSuccess(session, blocks);
    delete[] buffer;
    return true;
  }
  return false;
}

void EndpointImpl::InnerCheckFrameComplete(Session &session,
                                           std::optional<uint16_t> seq) {
  bool is_order = session.param.is_order;
  bool is_reliable = session.param.is_reliable;
  int64_t max_retrans_ms = session.param.max_retrans_time;
  uint64_t now = Time::NowMs();
  // 可靠只有收到新包的时候才检查
  if (seq == std::nullopt && is_reliable) return;
  while (1) {
    if (is_order && is_reliable) {
      if (session.recv_cache.begin()->first == session.recv_seq) {
        if (InnerCallBackCompleteFrame(session, session.recv_seq)) continue;
      }
      break;
    } else if (is_order && !is_reliable) {
      if (seq != std::nullopt &&
          SeqSortWarper<uint16_t>()(*seq, session.recv_seq))
        InnerCallBackCompleteFrame(session, seq);
      break;
    } else if (!is_order && is_reliable) {
      InnerCallBackCompleteFrame(session, seq);
      break;
    } else {
      InnerCallBackCompleteFrame(session, seq);
    }
  }
}

bool EndpointImpl::VerifyStatus(FrameType type) {
  switch (this->status) {
    case Status::Init: {
      return type == FrameType::ShakeHands;
    }
    case Status::Connecting: {
      return type == FrameType::ShakeHands ||
             type == FrameType::ShakeHandsAck || type == FrameType::KeepAlive ||
             type == FrameType::Data || type == FrameType::ReTransData;
    }
    case Status::Connected: {
      return type != FrameType::ShakeHandsAck;
    }
    case Status::DisConnecting: {
      return type == FrameType::KeepAlive || type == FrameType::DisConnect ||
             type == FrameType::DisConnectAck;
    }
    case Status::DisConnected: {
      return type == FrameType::DisConnect || type == FrameType::DisConnectAck;
    }
    case Status::TimeOut: {
        return type == FrameType::KeepAlive || type == FrameType::KeepAliveAck || type == FrameType::DisConnect || type == FrameType::DisConnectAck;
    }
    default:
      return false;
  }
}

void EndpointImpl::InnerSendTo(const uint8_t *data, size_t len,
                               const SocketAddress &remote) {
  if (this->configure.channel)
    this->configure.channel->Send((uint8_t *)data, len);
  else
    this->socket->SendTo((char *)data, len,
                         const_cast<SocketAddress &>(remote));
}

void EndpointImpl::InnerSend(Session &session) {
  switch (session.status) {
    case Session::Status::None: {
      Block *block = session.send_cache.begin()->second.get();
      block->frame.header.ext = true;
      size_t off_set = 1 + sizeof(struct SessionConfigure);
      block->frame.len += off_set;
      memmove(block->frame.data + off_set, block->frame.data,
              block->frame.payload_len);
      block->frame.data[0] = 1;
      memcpy(block->frame.data + 1, &session.configure,
             sizeof(struct SessionConfigure));
      session.status = Session::Status::Connecting;
      InnerSendBlock(session, block);
      break;
    }
    case Session::Status::Connecting: {
      Block *block = session.send_cache.begin()->second.get();
      PandaCheck(block->frame.header.ext == true);
      InnerSendBlock(session, block);
      break;
    }
    case Session::Status::Connected: {
      bool has_first_send = false;
      for (BlockRIterm it = session.send_cache.rbegin();
           it != session.send_cache.rend(); ++it) {
        if (it->second->retrans_count == 0) {
          has_first_send = true;
          LOG(Debug) << "Session:" << session.param.sid << " Retrans "
                     << it->first << " For " << it->second->retrans_count++
                     << " Time";
          InnerSendBlock(session, it->second.get());
        } else {
          if (session.param.enable_time_out_retrans &&
              it->second->acked == false &&
              NowUint32() - ByteReaderWriter<uint32_t>::ReadByte(
                                it->second->frame.header.ts) >
                  it->second->retrans_count * rtt) {
            LOG(Debug) << "Session:" << session.param.sid << " Retrans "
                       << it->first << " For " << it->second->retrans_count++
                       << " Time";
            InnerSendBlock(session, it->second.get());
          } else if (has_first_send) {
            break;
          }
        }
      }
      break;
    }
    default:
      break;
  }
}

void EndpointImpl::InnerRecvSack(Session &session, char *data, size_t len) {
  Frame *frame = reinterpret_cast<Frame *>(data);
  uint16_t seq = ByteReaderWriter<uint16_t>::ReadByte(frame->header.seq);
  uint32_t ts = ByteReaderWriter<uint32_t>::ReadByte(frame->header.ts);
  size_t bit_map_len = len - sizeof(Header);
  size_t payload_offset = 0;
  if (frame->header.ext) {
    payload_offset = 1;
    for (int i = 0; i < frame->data[0]; i++) {
      struct ExpandBase *ext_base =
          reinterpret_cast<struct ExpandBase *>(frame->data + payload_offset);
      payload_offset += ext_base->len;
    }
    bit_map_len -= payload_offset;
  }
  if (SeqSortWarper<uint16_t>()(seq, session.last_recv_sack_seq)) return;

  BlockIterm it = session.send_cache.find(seq);
  if (it != session.send_cache.begin()) {
    size_t erase_cache = 0;
    for (auto it_erase = session.send_cache.begin(); it_erase != it;
         ++it_erase) {
      erase_cache += it_erase->second->frame.payload_len;
    }
    LOG(Debug) << "Session:" << session.param.sid
               << " Clear Send Cache:" << session.send_cache.begin()->first
               << "-" << seq << " Size:" << erase_cache;
    session.send_cache.erase(session.send_cache.begin(), it);
    PandaCheck(session.send_cache_size >= erase_cache)
        << " Session:" << session.param.sid << " EraseSize:" << erase_cache
        << " SendCache:" << session.send_cache_size;
    session.send_cache_size -= erase_cache;
    it = session.send_cache.begin();
  }

  if (seq == session.send_seq && session.send_cache.empty() == false) {
    size_t erase_cache = 0;
    for (auto &it : session.send_cache) {
      erase_cache += it.second->frame.payload_len;
    }
    LOG(Debug) << "Session:" << session.param.sid
               << " Clear Send Cache:" << session.send_cache.begin()->first
               << "-" << session.send_cache.rbegin()->first
               << " Size:" << erase_cache;
    session.send_cache_size -= erase_cache;
    session.send_cache.clear();
    return;
  }

  if (it == session.send_cache.end()) return;

  for (; it != session.send_cache.end(); ++it) {
    if (it == session.send_cache.begin()) {
      if (SeqSortWarper<uint32_t>()(ts, ByteReaderWriter<uint32_t>::ReadByte(
                                            it->second->frame.header.ts)) ==
          false) {
        LOG(Debug) << "Session:" << session.param.sid
                   << " Recv Sack Retrans:" << it->first
                   << " Time:" << it->second->retrans_count++;
        InnerSendBlock(session, it->second.get());
      }
    } else if (it->second->acked == false) {
      uint16_t diff = it->first - seq - 1;
      size_t index = diff / CHAR_BIT + 1;
      size_t off_set = diff % CHAR_BIT;
      if (index < bit_map_len) {
        if (frame->data[payload_offset + index] &
            (1 << (CHAR_BIT - 1 - off_set))) {
          it->second->acked = true;
          LOG(Debug) << "Session:" << session.param.sid
                     << " Recv Sack Packet:" << it->first << " Remote Acked";
        } else if (SeqSortWarper<uint32_t>()(
                       ts, ByteReaderWriter<uint32_t>::ReadByte(
                               it->second->frame.header.ts)) == false) {
          LOG(Debug) << "Session:" << session.param.sid
                     << " Recv Sack Retrans:" << it->first
                     << " Time:" << it->second->retrans_count++;
          InnerSendBlock(session, it->second.get());
        }
      } else {
        break;
      }
    }
  }
}

void EndpointImpl::InnerRecv(Session &session, char *data, size_t len) {
  Frame *frame = reinterpret_cast<Frame *>(data);
  uint16_t seq = ByteReaderWriter<uint16_t>::ReadByte(frame->header.seq);

  if (SeqSortWarper<uint16_t>()(seq, session.recv_seq)) return;

  auto it = session.recv_cache.find(seq);
  if (it == session.recv_cache.end()) {
    std::unique_ptr<Block> block(new Block());
    block->called = false;
    block->recv_time_ms = Time::NowMs();
    size_t offset = 0;
    if (frame->header.ext) {
      offset = 1;
      for (int i = 0; i < frame->data[0]; i++) {
        struct ExpandBase *ext_base =
            reinterpret_cast<struct ExpandBase *>(frame->data + offset);
        offset += ext_base->len;
      }
    }
    // 清除拓展
    block->frame.payload_len = len - sizeof(Header) - offset;
    memcpy(block->frame.header_data, frame->header_data, sizeof(Header));
    memcpy(block->frame.data, frame->data + offset, block->frame.payload_len);
    block->frame.header.ext = 0;
    block->frame.len = len - offset;
    if (session.recv_cache.emplace(std::make_pair(seq, std::move(block)))
            .second) {
      session.sack_generator.receivePacket(seq);
      session.recv_data_num_since_last_sack++;
      InnerCheckFrameComplete(session, seq);
    }
  }
  InnerSendSack(session);
}

void EndpointImpl::InnerSendSack(Session &session) {
  uint64_t now = Time::NowMs();
  if (now - session.last_sack_send_ms >= sack_max_interval ||
      (session.recv_data_num_since_last_sack > sack_interval_large_packet &&
       now - session.last_sack_send_ms >= sack_min_interval)) {
    session.recv_data_num_since_last_sack = 0;
    Block sack;
    sack.frame.header.ver = version;
    sack.frame.header.type = FrameType::Sack;
    sack.frame.header.ext = 0;
    sack.frame.header.pos = 0;
    sack.frame.header.padding = 0;
    ByteReaderWriter<uint32_t>::WriteByte(sack.frame.header.sid,
                                          session.param.sid);
    ByteReaderWriter<uint16_t>::WriteByte(
        sack.frame.header.seq, session.sack_generator.getMinSeqNum());
    uint32_t ts = this->jitter_estimator.max_packet_ts -
                  (uint32_t)(now - this->jitter_estimator.max_packet_ts_ms) -
                  this->jitter_estimator.jitter_histogram.Quantile(kQuantile);
    ByteReaderWriter<uint32_t>::WriteByte(sack.frame.header.ts, ts);
    const std::vector<uint8_t> &bit_map = session.sack_generator.getBitmap();
    sack.frame.payload_len =
        bit_map.size() > sack_max_size ? sack_max_size : bit_map.size();
    memcpy(sack.frame.data, bit_map.data(), sack.frame.payload_len);
    sack.frame.len = sack.frame.payload_len + sizeof(Header);
    LOG(Debug) << "Session:" << session.param.sid
               << " SackSend:" << session.sack_generator.getMinSeqNum()
               << " Ts:" << ts
               << " BitMap:" << session.sack_generator.ToString();
    InnerSendBlock(session, &sack);
  }
}

void EndpointImpl::Close() {
  workerThread->Invoke(PandaFromHere, [&]() {
    PandaCheck(status == Status::Connected);
    status = Status::DisConnecting;
    InnerSendControlFrame(FrameType::DisConnect);
  });
}

void EndpointImpl::OnReadFrom(char *data, size_t len,
                              const SocketAddress &from) {
  if (this->status != Status::Init && !(*this->remoteAddr == from)) {
    return;
  }
  Header *header = reinterpret_cast<Header *>(data);
  Frame *frame = reinterpret_cast<Frame *>(data);
  if (VerifyStatus(static_cast<FrameType>(header->type)) == false) {
    LOG(Debug) << "Transport:" << this << " Status:" << this->status
               << " FrameType:" << header->type << " Verify Failed";
    return;
  }
  uint32_t ts = ByteReaderWriter<uint32_t>::ReadByte(header->ts);
  uint32_t sid = ByteReaderWriter<uint32_t>::ReadByte(header->sid);
  uint64_t now = Time::NowMs();

  if (CalculateRtt((FrameType)header->type)) {
    this->rtt = NowUint32() - ts;
  } else {
    InnerUpdateRecvJitter(ts, now);
  }


  LOG(Debug) << "Transport:" << this << " Status:" << this->status << " Sid:" << sid << " rtt:" << rtt
      << " FrameType:" << (int)(header->type) << " ts:" << ts << " Seq:" << ByteReaderWriter<uint32_t>::ReadByte(header->seq) << " GlobalSeq:" << ByteReaderWriter<uint32_t>::ReadByte(header->g_seq);
  this->globalSeqStatistician.UpdateSeq(
      ByteReaderWriter<uint32_t>::ReadByte(header->g_seq));
  switch (header->type) {
    case FrameType::ShakeHands:
    case FrameType::ShakeHandsAck: {
      this->InnerRecvShakeHands(static_cast<FrameType>(header->type), from, now,
                                ts);
      break;
    }
    case FrameType::KeepAlive:
    case FrameType::KeepAliveAck: {
      this->InnerRecvKeepAlive(static_cast<FrameType>(header->type), now, ts);
      break;
    }
    case FrameType::DisConnect:
    case FrameType::DisConnectAck: {
      this->InnerRecvDisConnect(static_cast<FrameType>(header->type), now, ts);
      break;
    }
    case FrameType::Data:
    case FrameType::ReTransData: {
      auto it = this->session.find(sid);
      if (it == this->session.end() && header->type == FrameType::Data &&
          header->ext) {
        uint8_t ext_num = frame->data[0];
        size_t off_set = 1;
        struct SessionConfigure *configure = nullptr;
        for (int i = 0; i < ext_num; i++) {
          struct ExpandBase *ext_base =
              reinterpret_cast<struct ExpandBase *>(frame->data + off_set);
          if (ext_base->type == ExpandType::SessionConfigure) {
            configure = reinterpret_cast<struct SessionConfigure *>(ext_base);
            break;
          }
          off_set += ext_base->len;
        }
        if (configure) {
          it = this->session.emplace(std::make_pair(sid, Session(configure)))
                   .first;

          LOG(Debug) << "Session:" << it->second.param.sid
                     << " Create: is_order:" << it->second.param.is_order
                     << " is_reliable:" << it->second.param.is_reliable
                     << " max_retrans_num:" << it->second.param.max_retrans_num
                     << " max_retrans_time:"
                     << it->second.param.max_retrans_time;
          it->second.status = Session::Status::Connected;
          this->configure.listener->OnSessionConnect(this, it->second.param);
        }
      }

      if (it != this->session.end()) {
        InnerRecv(it->second, data, len);
      }
      /* code */
      break;
    }
    case FrameType::Sack: {
      auto it = this->session.find(sid);
      if (it != this->session.end()) {
        if (it->second.status == Session::Connecting) {
            LOG(Debug) << "Transport:" << this << " Session:" << sid << " Recv Sack Connected";
            it->second.status = Session::Connected;
        }
        InnerRecvSack(it->second, data, len);
      }
      else {
        LOG(Debug) << "Transport:" << this << " Session:" << sid << " Can`t Find Session";
      }
      break;
    }

    default:
      break;
  }
}

void EndpointImpl::OnSend(char *data, size_t len, int status, const char *msg) {
  if (status != 0) {
    LOG(Error) << "Transport:" << this << " Send Failed:" << status
               << " Error:" << msg;
  }
}

void EndpointImpl::InnerTimer() {
  uint64_t now = Time::NowMs();
  switch (status) {
    case EndpointImpl::Init:
    case EndpointImpl::DisConnected:
      break;
    case EndpointImpl::Connecting: {
      if (now > this->last_keepalive_send + configure.ticket_ms) {
        InnerSendControlFrame(FrameType::ShakeHands);
      }
      break;
    }
    case EndpointImpl::TimeOut: {
      if (now > this->last_keepalive_send + configure.ticket_ms) {
        InnerSendControlFrame(FrameType::KeepAlive);
      }
      break;
    }
    case EndpointImpl::Connected: {
      if (now > this->last_keepalive_send + keep_alive_interval) {
        InnerSendControlFrame(FrameType::KeepAlive);
      }
      for (auto &it : this->session) {
        InnerSend(it.second);
      }
      break;
    }
    case EndpointImpl::DisConnecting: {
      InnerSendControlFrame(FrameType::DisConnect);
      break;
    }
    default:
      break;
  }
  workerThread->PostDelayTask(
      PandaFromHere, [this]() { this->InnerTimer(); }, configure.ticket_ms);
}

void EndpointImpl::InnerSendControlFrame(FrameType type, uint32_t ts) {
  uint64_t nowMs = Time::NowMs();
  if (this->last_recv_ack_ms + time_out_ms < nowMs &&
      (this->status != Status::TimeOut && this->status == Status::Connected)) {
    this->status = Status::TimeOut;
    LOG(Warning) << "Transport:" << this << " More Than " << time_out_ms
                 << " ms No Recv Ack Status Change To TimeOut";
  }
  Frame frame;
  frame.header.ver = version;
  frame.header.type = type;
  frame.header.ext = 0;
  frame.header.pos = 0;
  frame.header.padding = 0;
  ByteReaderWriter<uint32_t>::WriteByte(frame.header.sid, 0);
  ByteReaderWriter<uint16_t>::WriteByte(frame.header.seq, 0);
  ByteReaderWriter<uint32_t>::WriteByte(frame.header.g_seq, globalSeq++);
  ByteReaderWriter<uint32_t>::WriteByte(frame.header.ts, ts);
  frame.len = sizeof(struct Header);
  if (type == FrameType::KeepAlive || type == FrameType::ShakeHands) {
    this->last_keepalive_send = nowMs;
  }

  if (this->configure.channel)
    this->configure.channel->Send((uint8_t *)frame.header_data, frame.len);
  else
    this->socket->SendTo((char *)frame.header_data, frame.len,
                         *this->remoteAddr);
}

void EndpointImpl::InnerSendDisconnect() {}